package com.googlecode.lazyrecords.berkeleydb;

import com.googlecode.lazyrecords.AbstractRecords;
import com.googlecode.lazyrecords.Definition;
import com.googlecode.lazyrecords.Record;
import com.googlecode.lazyrecords.SourceRecord;
import com.googlecode.totallylazy.CloseableList;
import com.googlecode.totallylazy.Function;
import com.googlecode.totallylazy.Function1;
import com.googlecode.totallylazy.Option;
import com.googlecode.totallylazy.Pair;
import com.googlecode.totallylazy.Predicate;
import com.googlecode.totallylazy.Sequence;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.OperationStatus;

import java.io.Closeable;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.UUID;

import static com.googlecode.lazyrecords.Definition.methods.sortFields;
import static com.googlecode.totallylazy.Predicates.is;
import static com.googlecode.totallylazy.Sequences.repeat;
import static com.sleepycat.je.OperationStatus.SUCCESS;
import static java.util.UUID.randomUUID;

public class JeRecords extends AbstractRecords implements Closeable {
    private final CloseableList closeables = new CloseableList();
    private final Storage storage;
    private final Option<JeTransaction> transaction;

    public JeRecords(Storage storage, Option<JeTransaction> transaction) {
        this.storage = storage;
        this.transaction = transaction;
    }


    @Override
    public Sequence<Record> get(Definition definition) {
        return new JeSequence(closeables, storage.get(definition), transaction);
    }

    @Override
    public Number add(Definition definition, Sequence<Record> records) {
        RecordTuple recordTuple = new RecordTuple();
        return repeat(newKey()).zip(
                records.map(RecordTuple.toEntry(definition, recordTuple))).
                map(putInto(definition)).size();
    }

    private Function1<Pair<DatabaseEntry, DatabaseEntry>, OperationStatus> putInto(final Definition definition) {
        return new Function1<Pair<DatabaseEntry, DatabaseEntry>, OperationStatus>() {
            @Override
            public OperationStatus call(Pair<DatabaseEntry, DatabaseEntry> pair) throws Exception {
                return storage.get(definition).primary().put(JeTransaction.transaction(transaction), pair.first(), pair.second());
            }
        };
    }

    private Function<DatabaseEntry> newKey() {
        return new Function<DatabaseEntry>() {
            @Override
            public DatabaseEntry call() throws Exception {
                return key(randomUUID());
            }
        };
    }

    private DatabaseEntry key(UUID id) throws UnsupportedEncodingException {
        return new DatabaseEntry(id.toString().getBytes("UTF-8"));
    }

    @Override
    public Number remove(Definition definition, Predicate<? super Record> predicate) {
        return get(definition).filter(predicate).realise().
                map(uuid()).
                map(delete(definition)).
                filter(is(SUCCESS)).size();
    }

    private Function1<Record, UUID> uuid() {
        return new Function1<Record, UUID>() {
            @Override
            @SuppressWarnings("unchecked")
            public UUID call(Record record) throws Exception {
                return ((SourceRecord<UUID>) record).value();
            }
        };
    }

    private Function1<UUID, OperationStatus> delete(final Definition definition) {
        return new Function1<UUID, OperationStatus>() {
            @Override
            public OperationStatus call(UUID uuid) throws Exception {
                return storage.get(definition).primary().delete(JeTransaction.transaction(transaction), key(uuid));
            }
        };
    }

    @Override
    public void close() throws IOException {
        closeables.close();
    }
}
